package com.toutiao.nssue


import scala.collection.mutable.ArrayBuffer

import org.elasticsearch.common.settings.{Settings,ImmutableSettings}
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.index.query.FilterBuilders._
import org.elasticsearch.index.query.QueryBuilders._
import org.elasticsearch.action.search.SearchType
import org.elasticsearch.common.unit.TimeValue
import org.elasticsearch.search.aggregations.metrics.max.Max
import org.elasticsearch.search.aggregations.AggregationBuilders.max

import net.liftweb.json.{JsonAST,Extraction,Printer,parse}
import net.liftweb.json.JsonDSL._

import scala.collection.JavaConverters._

import java.util.Date
import java.sql.Timestamp

import org.apache.log4j.Logger

import com.toutiao.score.Topic
import com.toutiao.issue.SimpleDate

class IssueStore(
		esName: String,
		esUri: String,
		esPort: Int,
		esIndexName: String,
		esTypeName: String    
    ) {

  
  val _log = Logger.getLogger(this.getClass.getName)
  _log.info("esName "+esName)
  _log.info("esUri "+esUri)
  _log.info("esPort "+esPort)
  _log.info("esIndex "+esIndexName)
  _log.info("esType "+esTypeName)
  
  private
  val s = ImmutableSettings.settingsBuilder().
  			put("cluster.name",esName).build()
  			
  private
  val ES = new TransportClient(s).addTransportAddress(
		  	new InetSocketTransportAddress(esUri,esPort))
		  	
  implicit val _ = net.liftweb.json.DefaultFormats
  
  private
  def conv(n:Topic) = {
    val vs = Printer.compact(JsonAST.render(Extraction.decompose(n.v)))
    val ct = SimpleDate.format(n.c)
    val ut = SimpleDate.format(n.u)
    val ms = n.m("m").toList
    val name = n.i
    val cnt = ms.size
    
    val z = ("i"->name)~("c"->ct)~("u"->ut)~("v"->vs)~("m"->ms)~("n"->cnt)
    z
  }
  
  def insert(e:Topic) {
    val z = conv(e)
    val i = e.i
    val line = Printer.compact(JsonAST.render(z))
    ES.prepareIndex(esIndexName,esTypeName,i).setSource(line).execute.actionGet
  }
  
  def insert(ns:Array[Topic]) = {
    if (ns.size > 0){
      val bulkRequest = ES.prepareBulk
      ns.foreach(nu => {
        val z = conv(nu)
        val i = nu.i
        val line = Printer.compact(JsonAST.render(z))
        
        val m = ES.prepareIndex(esIndexName,esTypeName,i).setSource(line)
        bulkRequest.add(m)
      })
      bulkRequest.execute.actionGet
    } else {
      
    }
  }
  
  def getMaxTime(tt:String) = {
    val resp = ES.prepareSearch(esIndexName).setTypes(esTypeName)
    			.addAggregation(max("maxTime").field(tt)).execute.actionGet
    val r:Max = resp.getAggregations.get("maxTime")
    val x = r.getValue
    val z = new Date(new Timestamp(x.toLong).getTime)
    val t = SimpleDate.increment(z, -8*60) // 这里不知道会不会有问题
    t
  }
  
  def retrieve(st:Date, et:Date) = {
    val rt = "c"
    
    val xs = new ArrayBuffer[Topic]
	  var scrollResp = ES.prepareSearch(esIndexName).
						setTypes(esTypeName).
						setSearchType(SearchType.SCAN).
						setScroll(new TimeValue(60000)).
						setPostFilter(rangeFilter(rt).
						    from(SimpleDate.format(st)).
						    to(SimpleDate.format(et)).
						    includeLower(true).
						    includeUpper(false)).
						setSize(200).
						execute().actionGet()
						
	var z = 1
	while (z > 0){
	  scrollResp = ES.prepareSearchScroll(scrollResp.getScrollId).
	  					setScroll(new TimeValue(60000)).
	  					execute.actionGet
	  for (n <- scrollResp.getHits.hits) {
	    val s = n.getSource
	    val name = s.get("i").toString
	    val ct = SimpleDate.parse(s.get("c").asInstanceOf[String])
	    val ut = SimpleDate.parse(s.get("u").asInstanceOf[String])
	    val vs = parse(s.get("v").asInstanceOf[String]).extract[Map[String,Double]]
	    val ms = s.get("cluster").asInstanceOf[java.util.ArrayList[String]].asScala.toArray
	    
	    val e = new Topic
	    e.i = name
	    e.c = ct
	    e.u = ut
	    e.v = vs
	    e.m = Map("m"->ms)
	    e.s = 1.0
	    xs += e
	    //xs += new Event(name,ms,vs,ct,ut,1.0)
	  }
	  
	  if (scrollResp.getHits.hits.length == 0)
	    z = 0
	}
    _log.info("Got "+xs.size)
    xs.toArray.sortWith(_.c.getTime < _.c.getTime)
  }
  
  def close = ES.close
}
